package com.sunday.common.sharding.sphere.jdbc.leaf.algorithm.sharding.interval;

import com.google.common.base.Preconditions;
import com.google.common.collect.Range;
import com.sunday.common.core.constants.StringConstants;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.shardingsphere.sharding.api.sharding.complex.ComplexKeysShardingAlgorithm;
import org.apache.shardingsphere.sharding.api.sharding.complex.ComplexKeysShardingValue;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * 基于Leaf的区间范围分片算法。
 * 参考 : org.apache.shardingsphere.sharding.cosid.algorithm.sharding.interval.AbstractCosIdIntervalShardingAlgorithm
 * org.apache.shardingsphere.sharding.algorithm.sharding.complex.ComplexInlineShardingAlgorithm
 * '@Immutable' Immutable是“永恒的”“不会改变”的意思。在Immutable Patttern中，有着能够保证实例状态绝不会改变的类(immutable 类)。因为访问这个实例时，可以省去使用共享互斥机制所会浪费的时间，提高系统性能。java.lang.String就是一个Immutable的类。
 * 不可变对象永远不会发生改变，其字段的值只在构造函数运行时设置一次，其后就不会再改变。例如JDK中常见的两种基本数据类型String和Integer，它们都是不可变对象。为了理解immutable与mutable的区别
 * <p>
 * 自定义分片算法
 * 提供接口让应用开发者自行实现与业务实现紧密相关的分片算法，并允许使用者自行管理真实表的物理分布。 自定义分片算法又分为：
 * <p>
 * 标准分片算法
 * 用于处理使用单一键作为分片键的 =、IN、BETWEEN AND、>、<、>=、<= 进行分片的场景。
 * <p>
 * 复合分片算法
 * 用于处理使用多键作为分片键进行分片的场景，包含多个分片键的逻辑较复杂，需要应用开发者自行处理其中的复杂度。
 * <p>
 * Hint 分片算法
 * 用于处理使用 Hint 行分片的场景。
 *
 * @author zsy
 * @since 2022/12/29
 */
@Slf4j
public abstract class AbstractLeafIntervalShardingAlgorithm implements ComplexKeysShardingAlgorithm<Comparable<?>> {

    /**
     * 特别注意 下界值为包含关系，上界值不为包含，就是取前不取后
     */
    // 时间分片下界值单位，必须遵循 Java ChronoUnit 的枚举值。例如：MONTHS
    public static final String DATE_TIME_LOWER_UNIT_KEY = "datetime-lower-unit";
    // 时间分片下界值时间间隔，超过该时间间隔将返回 datetime-lower-unit 与 datetime-lower-amount 计算后的符合最小时间范围
    public static final String DATE_TIME_LOWER_AMOUNT_KEY = "datetime-lower-amount";
    // 时间分片上界值单位，必须遵循 Java ChronoUnit 的枚举值。例如：MONTHS
    public static final String DATE_TIME_UPPER_UNIT_KEY = "datetime-upper-unit";
    // 时间分片上界值时间间隔，超过该时间间隔将返回 datetime-upper-unit 与 datetime-upper-amount 计算后的符合最大时间范围
    public static final String DATE_TIME_UPPER_AMOUNT_KEY = "datetime-upper-amount";

    /**
     * 默认分界规则固定已 1 为单位，所以去除掉了 INTERVAL_AMOUNT_KEY 的设置
     * 因为当上限和下限不是固定值的时候，无法让操作者每次都记住上一个不定间隔的情况下是在什么时间刻度，而系统启动的不定性会使数据混乱
     * 所以采取固定间隔模式，让所有的刻度一致
     */
    // 分片数据源或真实表的后缀格式，必须遵循 Java DateTimeFormatter 的格式，必须和 datetime-interval-unit 保持一致。例如：yyyyMM
    public static final String SHARDING_SUFFIX_FORMAT_KEY = "sharding-suffix-pattern";
    // 分片键时间间隔单位，必须遵循 Java ChronoUnit 的枚举值。例如：MONTHS
    public static final String INTERVAL_UNIT_KEY = "datetime-interval-unit";
//    // 分片键时间间隔，超过该时间间隔将进入下一分片
//    public static final String INTERVAL_AMOUNT_KEY = "datetime-interval-amount";

    /**
     * 分片算法命中规则
     * 1. 优先精准分片，其次范围分片
     * 2. 优先雪花正则，其次时间正则
     * 3. 当前分支可能性为 2 * 2, 4中分支中优先命中执行完毕后立即返回
     */
    // 雪花正则
    public static final String LEAF_REGEX_KEY = "leaf-regex";
    // 时间正则
    public static final String TIME_REGEX_KEY = "time-regex";

    // 时区ID
    public static final String ZONE_ID_KEY = "zone-id";
    // 雪花正则
    public static final String LEAF_REGEX = "(.*_id$)|(id)";
    // 时间正则
    public static final String TIME_REGEX = "(.*_time$)|(.*_date$)";

    @Getter
    private Properties props;

    private ZoneId zoneId;

    private String leafRegex;

    private String timeRegex;

    private IntervalTimeline intervalTimeline;

    @Override
    public void init(final Properties props) {
        this.props = props;
        this.zoneId = getZoneId(props);
        this.leafRegex = getLeafRegex(props);
        this.timeRegex = getTimeRegex(props);
        this.intervalTimeline = getIntervalTimeline(props);
    }

    private ZoneId getZoneId(final Properties props) {
        return props.containsKey(ZONE_ID_KEY) ? ZoneId.of(props.getProperty(ZONE_ID_KEY)) : ZoneId.systemDefault();
    }

    private String getLeafRegex(final Properties props) {
        return props.containsKey(LEAF_REGEX_KEY) ? props.getProperty(LEAF_REGEX_KEY) : LEAF_REGEX;
    }

    private String getTimeRegex(final Properties props) {
        return props.containsKey(TIME_REGEX_KEY) ? props.getProperty(TIME_REGEX_KEY) : TIME_REGEX;
    }

    private String getRequiredValue(final Properties props, final String key) {
        Preconditions.checkArgument(props.containsKey(key), "%s can not be null.", key);
        return props.getProperty(key);
    }

    private IntervalTimeline getIntervalTimeline(final Properties props) {
        ChronoUnit lowerUnit = ChronoUnit.valueOf(getRequiredValue(props, DATE_TIME_LOWER_UNIT_KEY));
        int lowerAmount = Integer.parseInt(getRequiredValue(props, DATE_TIME_LOWER_AMOUNT_KEY));
        ChronoUnit upperUnit = ChronoUnit.valueOf(getRequiredValue(props, DATE_TIME_UPPER_UNIT_KEY));
        int upperAmount = Integer.parseInt(getRequiredValue(props, DATE_TIME_UPPER_AMOUNT_KEY));
        DateTimeFormatter suffixFormatter = DateTimeFormatter.ofPattern(getRequiredValue(props, SHARDING_SUFFIX_FORMAT_KEY));
        ChronoUnit stepUnit = ChronoUnit.valueOf(getRequiredValue(props, INTERVAL_UNIT_KEY));
        return new IntervalTimeline(suffixFormatter, stepUnit, Bound.of(lowerUnit, lowerAmount), Bound.of(upperUnit, upperAmount));
    }

    @Override
    public Collection<String> doSharding(final Collection<String> availableTargetNames, final ComplexKeysShardingValue<Comparable<?>> shardingValue) {
        String targetName = availableTargetNames.stream().findFirst().get();
        /**
         * 1. 优先精准分片，其次范围分片
         * =、IN
         */
        Map<String, Collection<Comparable<?>>> columnNameAndShardingValuesMap = shardingValue.getColumnNameAndShardingValuesMap();
        if (!columnNameAndShardingValuesMap.isEmpty()) {
            return precise(targetName, columnNameAndShardingValuesMap);
        }
        /**
         * BETWEEN AND、>、<、>=、<=
         */
        Map<String, Range<Comparable<?>>> columnNameAndRangeValuesMap = shardingValue.getColumnNameAndRangeValuesMap();
        if (!columnNameAndRangeValuesMap.isEmpty()) {
            return range(targetName, columnNameAndRangeValuesMap);
        }
        throw new UnsupportedOperationException("not valid sharding!");
    }

    /**
     * 选举
     * <p>
     * Map<String, Range<Comparable<?>>> columnNameAndRangeValuesMap
     * Map<String, Collection<Comparable<?>>> columnNameAndShardingValuesMap
     * <p>
     * 1. 当值存在时可参与选举
     * 2. 范围判定必须存在边界，否则不入选
     * 3. 优先雪花正则，其次时间正则
     * 4. 时间正则优先第一个命中者
     *
     * @param valuesMap
     * @return {@link Map.Entry< String,T>}
     */
    private <T> Map.Entry<String, T> pick(final Map<String, T> valuesMap) {
        Map.Entry<String, T> first = null;
        for (Map.Entry<String, T> entry : valuesMap.entrySet()) {
            if (!isValid(entry.getValue())) {
                continue;
            }
            if (entry.getKey().matches(leafRegex)) {
                return entry;
            }
            if (first == null && entry.getKey().matches(timeRegex)) {
                first = entry;
            }
        }
        return first;
    }

    /**
     * 是否有效
     * 1. 值不存在 无效
     * 2. 当值为 Range 类型必须存在有效边界
     *
     * @param shardingValue
     * @return {@link boolean}
     */
    private <T> boolean isValid(T shardingValue) {
        if (ObjectUtils.isEmpty(shardingValue)) {
            return false;
        }
        if (shardingValue instanceof Range range) {
            return range.hasLowerBound() && range.hasUpperBound();
        }
        return true;
    }

    /**
     * 精准分片 =、IN
     *
     * @param targetName
     * @param columnNameAndShardingValuesMap
     * @return {@link Collection< String>}
     */
    private Collection<String> precise(final String targetName, final Map<String, Collection<Comparable<?>>> columnNameAndShardingValuesMap) {
        return Optional.ofNullable(pick(columnNameAndShardingValuesMap)).map(entry -> {
            if (entry.getKey().matches(leafRegex)) {
                return entry.getValue().stream().map(shardingValue -> targetName + StringConstants.UNDERSCORE + intervalTimeline.sharding(snowflakeShardingValue(shardingValue))).collect(Collectors.toSet());
            } else {
                return entry.getValue().stream().map(shardingValue -> targetName + StringConstants.UNDERSCORE + intervalTimeline.sharding(dateTimeShardingValue(shardingValue))).collect(Collectors.toSet());
            }
        }).orElseThrow(() -> new UnsupportedOperationException("not valid sharding!"));
    }

    /**
     * 范围分片 BETWEEN AND、>、<、>=、<=
     *
     * @param targetName
     * @param columnNameAndRangeValuesMap
     * @return {@link Collection< String>}
     */
    private Collection<String> range(final String targetName, final Map<String, Range<Comparable<?>>> columnNameAndRangeValuesMap) {
        return Optional.ofNullable(pick(columnNameAndRangeValuesMap)).map(entry -> {
            Range<Comparable<?>> range = entry.getValue();
            Range<LocalDateTime> dateTimeRange;
            if (entry.getKey().matches(leafRegex)) {
                dateTimeRange = Range.range(snowflakeShardingValue(range.lowerEndpoint()), range.lowerBoundType(), snowflakeShardingValue(range.upperEndpoint()), range.upperBoundType());
            } else {
                dateTimeRange = Range.range(dateTimeShardingValue(range.lowerEndpoint()), range.lowerBoundType(), dateTimeShardingValue(range.upperEndpoint()), range.upperBoundType());
            }
            return intervalTimeline.sharding(dateTimeRange).stream().map(suffix -> targetName + StringConstants.UNDERSCORE + suffix).collect(Collectors.toSet());
        }).orElseThrow(() -> new UnsupportedOperationException("not valid sharding!"));
    }

    protected abstract LocalDateTime snowflakeShardingValue(Comparable<?> shardingValue);

    protected abstract LocalDateTime dateTimeShardingValue(Comparable<?> shardingValue);

}
